package cn.sczc.jh.client.hbase;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;

import org.apache.hadoop.hbase.CompareOperator;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.BufferedMutator;
import org.apache.hadoop.hbase.client.BufferedMutatorParams;
import org.apache.hadoop.hbase.client.ColumnFamilyDescriptor;
import org.apache.hadoop.hbase.client.ColumnFamilyDescriptorBuilder;
import org.apache.hadoop.hbase.client.Delete;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.HBaseAdmin;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.client.Table;
import org.apache.hadoop.hbase.client.TableDescriptor;
import org.apache.hadoop.hbase.client.TableDescriptorBuilder;
import org.apache.hadoop.hbase.filter.ByteArrayComparable;
import org.apache.hadoop.hbase.filter.ColumnPrefixFilter;
import org.apache.hadoop.hbase.filter.Filter;
import org.apache.hadoop.hbase.filter.FilterList;
import org.apache.hadoop.hbase.filter.PageFilter;
import org.apache.hadoop.hbase.filter.RowFilter;
import org.apache.hadoop.hbase.filter.SingleColumnValueFilter;
import org.apache.hadoop.hbase.util.Bytes;

/**
 * hbase封装工具
 */
@SuppressWarnings("deprecation")
public class HbaseUtils {

	/**
	 * 创建hbase表
	 *
	 * @param tableName 表名称
	 * @param cfs       列族的数组
	 * @return 是否创建成功
	 * @throws IOException
	 */
	public static boolean createTable(String tableName, String[] cfs) throws IOException {
		HBaseAdmin admin = (HBaseAdmin) HbaseConn.getHbaseConn().getAdmin();
		TableName name = TableName.valueOf(tableName);
		if (!admin.isTableAvailable(name)) {
			// 表描述器构造器
			TableDescriptorBuilder tableDescriptorBuilder = TableDescriptorBuilder.newBuilder(name);
			List<ColumnFamilyDescriptor> columnFamilyDescriptorList = new ArrayList<>();
			for (String columnFamily : cfs) {
				// 列族描述起构造器
				ColumnFamilyDescriptorBuilder columnFamilyDescriptorBuilder = ColumnFamilyDescriptorBuilder
						.newBuilder(Bytes.toBytes(columnFamily));
				// 获得列描述
				ColumnFamilyDescriptor columnFamilyDescriptor = columnFamilyDescriptorBuilder.build();
				columnFamilyDescriptorList.add(columnFamilyDescriptor);
			}
			// 设置列簇
			tableDescriptorBuilder.setColumnFamilies(columnFamilyDescriptorList);
			// 获得表描述器
			TableDescriptor tableDescriptor = tableDescriptorBuilder.build();
			// 创建表
			admin.createTable(tableDescriptor);
		}
		return true;
	}

	/**
	 * 禁用表
	 * 
	 * @param tableName
	 * @throws IOException
	 */
	public static void disableTable(String tableName) throws IOException {
		HBaseAdmin admin = (HBaseAdmin) HbaseConn.getHbaseConn().getAdmin();
		TableName name = TableName.valueOf(tableName);
		if (!admin.isTableDisabled(name)) {
			admin.disableTable(name);
		}
	}

	/**
	 * 清空表
	 * 
	 * @param tableName
	 */
	public static void truncate(String tableName) throws IOException {
		HBaseAdmin admin = (HBaseAdmin) HbaseConn.getHbaseConn().getAdmin();
		TableName name = TableName.valueOf(tableName);
		disableTable(tableName);
		admin.truncateTable(name, true);
	}

	/**
	 * 删除表
	 *
	 * @param tableName 表名称
	 * @return 是否操作成功
	 * @throws IOException
	 */
	public static boolean deleteTable(String tableName) throws IOException {
		HBaseAdmin admin = (HBaseAdmin) HbaseConn.getHbaseConn().getAdmin();
		admin.disableTable(TableName.valueOf(tableName));
		admin.deleteTable(TableName.valueOf(tableName));
		return true;
	}

	/**
	 * 获取数据表列表
	 * 
	 * @return
	 * @throws IOException
	 */
	public static List<TableDescriptor> listTables() throws IOException {
		HBaseAdmin admin = (HBaseAdmin) HbaseConn.getHbaseConn().getAdmin();
		List<TableDescriptor> tableDescriptors = admin.listTableDescriptors();
		return tableDescriptors;
	}

	/**
	 * @param tableName 表名称
	 * @param rowKey    唯一标识
	 * @param cfName    列族名
	 * @param qualifier 列标识
	 * @param data      数据
	 * @return 是否操作成功
	 * @throws IOException
	 */
	public static boolean putRow(String tableName, String rowKey, String cfName, String qualifier, String data)
			throws IOException {
		Table table = HbaseConn.getTable(tableName);
		Put put = new Put(Bytes.toBytes(rowKey));
		put.addColumn(Bytes.toBytes(cfName), Bytes.toBytes(qualifier), Bytes.toBytes(data));
		table.put(put);
		return true;
	}

	/**
	 * 批量添加
	 *
	 * @param tableName 表名
	 * @param puts      put数组
	 * @return 是否操作成功
	 * @throws IOException
	 */
	public static boolean putRows(String tableName, List<Put> puts) throws IOException {
		BufferedMutatorParams params = new BufferedMutatorParams(TableName.valueOf(tableName));
		params.writeBufferSize(5 * 1024 * 1024);
		BufferedMutator mutator = HbaseConn.getHbaseConn().getBufferedMutator(params);
		// Table table = HbaseConn.getTable(tableName);
		try {
			mutator.mutate(puts);
			mutator.flush();
		} finally {
			mutator.close();
		}
		// table.put(puts);
		return true;
	}

	/**
	 * 插入一条记录
	 * 
	 * @param tableName
	 * @param rowKey
	 * @param columnFamily
	 * @param column
	 * @param value
	 * @throws IOException
	 */
	public static void insertOne(String tableName, String rowKey, String columnFamily, String column, String value)
			throws IOException {
		Put put = new Put(Bytes.toBytes(rowKey));
		// 下面三个分别为，列族，列名，列值
		put.addColumn(Bytes.toBytes(columnFamily), Bytes.toBytes(column), Bytes.toBytes(value));
		// 得到 table
		Table table = HbaseConn.getHbaseConn().getTable(TableName.valueOf(tableName));
		// 执行插入
		table.put(put);
	}

	/**
	 * 多行多列多值-单列簇
	 * 
	 * @param tableName
	 * @param columnFamily
	 * @param mapList
	 * @throws IOException
	 */
	public static void insertAll(String tableName, String columnFamily, List<Map<String, String>> mapList)
			throws IOException {
		Table table = HbaseConn.getHbaseConn().getTable(TableName.valueOf(tableName));
		List<Put> puts = new ArrayList<>();

		for (Map<String, String> map : mapList) {
			Put put = new Put(Bytes.toBytes(map.get("rowKey")));
			put.addColumn(Bytes.toBytes(columnFamily), Bytes.toBytes(map.get("column")),
					Bytes.toBytes(map.get("value")));
			puts.add(put);
		}
		table.put(puts);
	}

	/**
	 * 单行多列多值-单列簇
	 * 
	 * @param tableName
	 * @param rowKey
	 * @param columnFamily
	 * @param mapList
	 * @throws IOException
	 */
	public static void insertAll(String tableName, String rowKey, String columnFamily,
			List<Map<String, String>> mapList) throws IOException {
		Table table = HbaseConn.getHbaseConn().getTable(TableName.valueOf(tableName));
		List<Put> puts = new ArrayList<>();

		for (Map<String, String> map : mapList) {
			Put put = new Put(Bytes.toBytes(rowKey));
			put.addColumn(Bytes.toBytes(columnFamily), Bytes.toBytes(map.get("column")),
					Bytes.toBytes(map.get("value")));
			puts.add(put);
		}
		table.put(puts);
		// table.close();
	}

	/**
	 * 更新数据
	 * 
	 * @param tableName
	 * @param rowKey
	 * @param columnFamily
	 * @param column
	 * @param value
	 * @throws IOException
	 */

	public static void update(String tableName, String rowKey, String columnFamily, String column, String value)
			throws IOException {
		Table table = HbaseConn.getHbaseConn().getTable(TableName.valueOf(tableName));
		Put put = new Put(Bytes.toBytes(rowKey));
		put.addColumn(Bytes.toBytes(columnFamily), Bytes.toBytes(column), Bytes.toBytes(value));
		table.put(put);
		// table.close();
	}

	/**
	 * 删除单行单列
	 * 
	 * @param tableName
	 * @param rowKey
	 * @param columnFamily
	 * @param column
	 * @throws IOException
	 */
	public static void delete(String tableName, String rowKey, String columnFamily, String column) throws IOException {
		Table table = HbaseConn.getHbaseConn().getTable(TableName.valueOf(tableName));
		Delete delete = new Delete(Bytes.toBytes(rowKey));
		delete.addColumn(Bytes.toBytes(columnFamily), Bytes.toBytes(column));
		table.delete(delete);
	}

	/**
	 * 删除单行多列
	 * 
	 * @param tableName
	 * @param rowKey
	 * @param columnFamily
	 * @param columnList
	 * @throws IOException
	 */
	public static void delete(String tableName, String rowKey, String columnFamily, String... columnList)
			throws IOException {
		Table table = HbaseConn.getHbaseConn().getTable(TableName.valueOf(tableName));
		Delete delete = new Delete(Bytes.toBytes(rowKey));
		for (String column : columnList) {
			delete.addColumns(Bytes.toBytes(columnFamily), Bytes.toBytes(column));
		}
		table.delete(delete);
	}

	/**
	 * 删除单行单列簇
	 * 
	 * @param tableName
	 * @param rowKey
	 * @param columnFamily
	 * @throws IOException
	 */
	public static void delete(String tableName, String rowKey, String columnFamily) throws IOException {
		Table table = HbaseConn.getHbaseConn().getTable(TableName.valueOf(tableName));
		Delete delete = new Delete(Bytes.toBytes(rowKey));
		delete.addFamily(Bytes.toBytes(columnFamily));
		table.delete(delete);
	}

	/**
	 * 删除单行
	 * 
	 * @param tableName
	 * @param rowKey
	 * @throws IOException
	 */
	public static void delete(String tableName, String rowKey) throws IOException {
		Table table = HbaseConn.getHbaseConn().getTable(TableName.valueOf(tableName));
		Delete delete = new Delete(Bytes.toBytes(rowKey));
		table.delete(delete);
	}

	/**
	 * 删除多行
	 * 
	 * @param tableName
	 * @param rowKeyList
	 * @throws IOException
	 */
	public static void delete(String tableName, String... rowKeyList) throws IOException {
		Table table = HbaseConn.getHbaseConn().getTable(TableName.valueOf(tableName));
		ArrayList<Delete> deleteList = new ArrayList<>();
		for (String rowKey : rowKeyList) {
			Delete delete = new Delete(Bytes.toBytes(rowKey));
			deleteList.add(delete);
		}
		table.delete(deleteList);
	}

	/**
	 * 获取单条数据
	 *
	 * @param tableName 表名
	 * @param rowKey    唯一标识
	 * @return 查询结果
	 * @throws IOException
	 */
	public static Result getRow(String tableName, String rowKey) throws IOException {
		Table table = HbaseConn.getTable(tableName);
		Get get = new Get(Bytes.toBytes(rowKey));
		return table.get(get);
	}

	/**
	 * 根据过滤器查找具体数据
	 *
	 * @param tableName
	 * @param rowKey
	 * @param filterList
	 * @return
	 * @throws IOException
	 */
	public static Result getRow(String tableName, String rowKey, FilterList filterList) throws IOException {
		Table table = HbaseConn.getTable(tableName);
		Get get = new Get(Bytes.toBytes(rowKey));
		get.setFilter(filterList);
		return table.get(get);
	}

	/**
	 * 全表扫描
	 *
	 * @param tableName
	 * @return
	 * @throws IOException
	 */
	public static ResultScanner getScanner(String tableName) throws IOException {
		Table table = HbaseConn.getTable(tableName);
		Scan scan = new Scan();
		scan.setCaching(1000);
		return table.getScanner(scan);
	}

	/**
	 * 批量检索
	 *
	 * @param tableName   表名
	 * @param startRowKey 起始rowKey
	 * @param endRowKey   终止rowKey
	 * @return ResultScanner实例
	 * @throws IOException
	 */
	public static ResultScanner getScanner(String tableName, String startRowKey, String endRowKey) throws IOException {

		Table table = HbaseConn.getTable(tableName);
		Scan scan = new Scan();
		scan.setStartRow(Bytes.toBytes(startRowKey));
		scan.setStopRow(Bytes.toBytes(startRowKey));
		scan.setCaching(1000);
		return table.getScanner(scan);

	}

	/**
	 * 查询表
	 * 
	 * @param tableName
	 * @param rowKey
	 * @return
	 * @throws IOException
	 */
	public static Result select(String tableName, String rowKey) throws IOException {
		Table table = HbaseConn.getHbaseConn().getTable(TableName.valueOf(tableName));
		Get get = new Get(Bytes.toBytes(rowKey));
		Result result = table.get(get);
		return result;

	}

	/**
	 * 全表扫描
	 * 
	 * @param tableName
	 * @return
	 * @throws IOException
	 */
	public static ResultScanner scan(String tableName) throws IOException {
		Table table = HbaseConn.getHbaseConn().getTable(TableName.valueOf(tableName));
		Scan scan = new Scan();
		ResultScanner scanner = table.getScanner(scan);

		return scanner;
	}

	/**
	 * 全表扫描-列簇
	 * 
	 * @param tableName
	 * @param columnFamily
	 * @return
	 * @throws IOException
	 */
	public static ResultScanner scan(String tableName, String columnFamily) throws IOException {
		Table table = HbaseConn.getHbaseConn().getTable(TableName.valueOf(tableName));
		Scan scan = new Scan();
		scan.addFamily(Bytes.toBytes(columnFamily));
		ResultScanner scanner = table.getScanner(scan);

		return scanner;

	}

	/**
	 * 全表扫描-列
	 * 
	 * @param tableName
	 * @param columnFamily
	 * @param column
	 * @return
	 * @throws IOException
	 */
	public static ResultScanner scan(String tableName, String columnFamily, String column) throws IOException {
		Table table = HbaseConn.getHbaseConn().getTable(TableName.valueOf(tableName));
		Scan scan = new Scan();
		scan.addColumn(Bytes.toBytes(columnFamily), Bytes.toBytes(column));
		ResultScanner scanner = table.getScanner(scan);

		return scanner;
	}

	/**
	 * 全表扫描-过滤器
	 * 
	 * @param tableName
	 * @param filter
	 * @return
	 * @throws IOException
	 */
	public static ResultScanner scan(String tableName, Filter filter) throws IOException {

		Table table = HbaseConn.getHbaseConn().getTable(TableName.valueOf(tableName));
		Scan scan = new Scan();
		scan.setFilter(filter);
		ResultScanner scanner = table.getScanner(scan);

		return scanner;
	}

	/**
	 * 分页全表扫描-过滤器
	 * 
	 * @param tableName
	 * @param filter
	 * @return
	 * @throws IOException
	 */
	public static ResultScanner scan(String tableName, Filter filter, String startRowKey) throws IOException {

		Table table = HbaseConn.getHbaseConn().getTable(TableName.valueOf(tableName));
		Scan scan = new Scan();
		scan.setFilter(filter);
		scan.withStartRow(Bytes.toBytes(startRowKey));
		ResultScanner scanner = table.getScanner(scan);

		return scanner;
	}

	/**
	 * 获取分页过滤器
	 * 
	 * @param size
	 * @return
	 */
	public static Filter pagetFilter(long size) {

		return new PageFilter(size);

	}

	/**
	 * singleColumnValueFilter
	 * 
	 * @param columnFamily
	 * @param column
	 * @param compareOperator
	 * @param value
	 * @return
	 */
	public static Filter singleColumnValueFilter(String columnFamily, String column, CompareOperator compareOperator,
			String value) {
		return new SingleColumnValueFilter(Bytes.toBytes(columnFamily), Bytes.toBytes(column), compareOperator,
				Bytes.toBytes(value));
	}

	/**
	 * rowFilter
	 * 
	 * @param compareOperator
	 * @param rowComparator
	 * @return
	 */
	public static Filter rowFilter(CompareOperator compareOperator, ByteArrayComparable rowComparator) {
		return new RowFilter(compareOperator, rowComparator);
	}

	/**
	 * columnPrefixFilter
	 * 
	 * @param prefix
	 * @return
	 */
	public static Filter columnPrefixFilter(String prefix) {
		return new ColumnPrefixFilter(Bytes.toBytes(prefix));
	}

	/**
	 * 过滤器集合
	 * 
	 * @param filterList
	 * @return
	 */
	public static FilterList filterListPassAll(Filter... filterList) {
		FilterList list = new FilterList(FilterList.Operator.MUST_PASS_ALL);
		for (Filter filter : filterList) {
			list.addFilter(filter);
		}
		return list;
	}

	/**
	 * 过滤器集合
	 * 
	 * @param
	 * @return
	 */
	public static FilterList filterListPassOne(Filter... filterList) {
		FilterList list = new FilterList(FilterList.Operator.MUST_PASS_ONE);
		for (Filter filter : filterList) {
			list.addFilter(filter);
		}
		return list;
	}

	/**
	 * 批量检索
	 *
	 * @param tableName   表名
	 * @param startRowKey 起始rowKey
	 * @param endRowKey   终止rowKey
	 * @param filterList  过滤器
	 * @return ResultScanner 实例
	 * @throws IOException
	 */
	public static ResultScanner getScanner(String tableName, String startRowKey, String endRowKey,
			FilterList filterList) throws IOException {

		Table table = HbaseConn.getTable(tableName);
		Scan scan = new Scan();
		scan.setStartRow(Bytes.toBytes(startRowKey));
		scan.setStopRow(Bytes.toBytes(startRowKey));
		scan.setFilter(filterList);
		scan.setCaching(1000);
		return table.getScanner(scan);
	}

	/**
	 * 查询指定的某列
	 * 
	 * @param tableName
	 * @param rowKey
	 * @param familyName
	 * @param columnName
	 * @return
	 * @throws Exception
	 */
	public static Result getResult(String tableName, String rowKey, String familyName, String columnName)
			throws Exception {
		Get get = new Get(Bytes.toBytes(rowKey));
		Table table = HbaseConn.getTable(tableName);
		get.addColumn(Bytes.toBytes(familyName), Bytes.toBytes(columnName));
		Result result = table.get(get);
		return result;
	}

	/**
	 * 删除操作
	 *
	 * @param tableName 表名
	 * @param rowKey    rowKey
	 * @return 是否操作成功
	 * @throws IOException
	 */
	public static boolean deleteRow(String tableName, String rowKey) throws IOException {
		Table table = HbaseConn.getTable(tableName);
		Delete delete = new Delete(Bytes.toBytes(rowKey));
		table.delete(delete);
		return true;
	}

	/**
	 * 删除操作 列族
	 *
	 * @param tableName 表名
	 * @param cfName    rowKey
	 * @return 是否操作成功
	 * @throws IOException
	 */
	public static boolean deleteColumnFamily(String tableName, String cfName) throws IOException {
		HBaseAdmin admin = (HBaseAdmin) HbaseConn.getHbaseConn().getAdmin();
		admin.deleteColumnFamily(TableName.valueOf(tableName), Bytes.toBytes(cfName));
		return true;
	}

	/**
	 * 删除列
	 *
	 * @param tableName  表名
	 * @param rowKey     rowKey
	 * @param cfName     列族名称
	 * @param qualifiter 删除操作 列
	 * @return
	 * @throws IOException
	 */
	public static boolean deleteQualifiter(String tableName, String rowKey, String cfName, String qualifiter)
			throws IOException {
		Table table = HbaseConn.getTable(tableName);
		Delete delete = new Delete(Bytes.toBytes(rowKey));
		delete.addColumn(Bytes.toBytes(cfName), Bytes.toBytes(qualifiter));
		table.delete(delete);
		return true;
	}

	/**
	 * 查询指定单cell内容
	 * 
	 * @param tableName
	 * @param rowKey
	 * @param family
	 * @param col
	 * @return
	 * @throws IOException
	 */
	public static String getCellData(String tableName, String rowKey, String family, String col) throws IOException {
		Table table = HbaseConn.getTable(tableName);
		String result = null;
		Get get = new Get(rowKey.getBytes());
		if (!get.isCheckExistenceOnly()) {
			get.addColumn(Bytes.toBytes(family), Bytes.toBytes(col));
			Result res = table.get(get);
			byte[] resByte = res.getValue(Bytes.toBytes(family), Bytes.toBytes(col));
			result = Bytes.toString(resByte);
		}
		return result;
	}

	public static void main(String[] args) throws Exception {
//		Result result = HbaseUtils.getRow("sensor_data", "fffe3baa-a013-44c8-a0a9-eaffd3e38233");
//		System.out.println(result.toString());
//		Result result1 = HbaseUtils.getResult("sensor_data", "fffe3baa-a013-44c8-a0a9-eaffd3e38233", "property_info",
//				"wirelessMode");
//		System.out.println(result1.toString());
//		String val = HbaseUtils.getCellData("sensor_data", "fffe3baa-a013-44c8-a0a9-eaffd3e38233", "property_info",
//				"wirelessMode");
//		System.out.println(val);
		HbaseUtils.truncate("sensor_data");
	}
}
